package cn.tedu.charging.order.consumer;

import cn.tedu.charging.common.constant.MqttDestinationConst;
import cn.tedu.charging.common.pojo.message.ChargingCheckResultMessage;
import cn.tedu.charging.common.pojo.message.ChargingProgressMsg;
import cn.tedu.charging.order.serialize.MessageSerializer;
import cn.tedu.charging.order.service.ConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 回调函数类,需要重建成容器bean对象并且交给绑定一个mqttClient使用
 * 专门负责监听这个mqttClient的一举一动
 * 1. 连接成功
 * 2. 发送消息
 * 3. 断开连接
 * 4. 收到消息
 */
@Component
@Slf4j
public class MqttClientCallback implements MqttCallbackExtended {
    @Autowired
    private MessageSerializer messageSerializer;
    @Autowired
    private ConsumerService consumerService;
    @Autowired
    private RedisTemplate redisTemplate;
    /*
        这个方法是消费的入口
        方法作用: 当mqttClient连接对象订阅了某个主题,一旦这个主题中出现消息,就会通过这个方法将消息传递给客户端
        参数:
        String topic: 消息来源的topic名称
        MqttMessage message: 消息对象,包含了消息的所有内容
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("消息接收成功,消息:{},接收的主题:{}",message,topic);
        //判断消息来源,反序列化消息成为 消息对象调用业务实现
        if (MqttDestinationConst.GUN_CHECK_RESULT_TOPIC.equals(topic)){
            log.debug("来自设备自检结果消息,消息内容:{}",message);
            ChargingCheckResultMessage msg = (ChargingCheckResultMessage) messageSerializer.deserialize(message.getPayload(), ChargingCheckResultMessage.class);
            //1.实现抢锁逻辑
            //1.1抢锁 setnx string数据结构 从redisTemplate中获取操作对象
            ValueOperations valueOps = redisTemplate.opsForValue();
            //1.2抢锁 key值和共享业务数据有关 orderNo
            String lockKey="charging:order:consume:"+msg.getOrderNo();
            //1.3执行抢锁
            Boolean haveLock = false;
            Integer count=0;
            try{
                do{
                    //do做了,就说明要抢锁了,count++计数
                    count++;
                    //只允许最多抢三次,如果count++之后已经等于4就说明已经抢了三次了 如果count!=1 说明是多次抢锁了
                    if (count>1&&count<4){
                        //说明多次抢锁没达到上限 沉睡100在抢
                        Thread.sleep(100);
                    }else if (count>=4){
                        //超过抢锁上限,别在抢了,业务也不执行了
                        log.error("超过抢锁上限,别在抢了,业务也不执行了,orderNo:{}",msg.getOrderNo());
                        return;
                    }
                    haveLock = valueOps.setIfAbsent(lockKey, "", 5, TimeUnit.SECONDS);
                }while (!haveLock);
                consumerService.handleCheckResult(msg);
            }catch (Exception e){
                log.error("抢锁失败",e);
            }finally{
                redisTemplate.delete(lockKey);
            }
        }else if (MqttDestinationConst.CHARGING_PROGRESS_TOPIC.equals(topic)){
            //同步充电进度来处理
            log.debug("来自充电进度消息,消息内容:{}",message);
            ChargingProgressMsg msg= (ChargingProgressMsg) messageSerializer.deserialize(message.getPayload(), ChargingProgressMsg.class);
            //业务抽象
            consumerService.handleChargingProgress(msg);
        }
    }
    /*
        方法作用: 当mqttClient调用connect并且成功和emqx建立通信之后立即调用
        参数:
        boolean reconnect: 如果本次建立的连接是重新连接,reconnect就是true 第一次连接false
        String serverURI: 连接服务端address
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (reconnect){
            log.info("重新连接成功,broker地址:{}",serverURI);
        }else{
            log.info("首次连接成功,broker地址:{}",serverURI);
        }
    }
    /*
        方法作用: 当mqttClient调用disconnect方法断开连接后,或者由于网络波动,导致和服务端连接通信超时失败,回调就会调用
        这个方法
        参数:
        Throwable cause: 断开连接的原因
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.error("连接断开,原因:{}",cause);
    }

    /*
       方法作用: 当mqttClient发送消息的时候,只要消息成功发送到emqx,不管emqx是否能够接收到
       反馈接收成功还是失败,都会调用这个deliveryComplete,处理发送完成的业务
       qos服务质量级别中0 表示根本不关心投递发送结果,所以,如果客户端mqttClient使用qos=0级别发送消息,不会有任何异常
       不会回调这个方法
       参数:
       IMqttDeliveryToken token: 发送的消息对象 映射,包含了你发送消息的所有内容,以及双发通信的结果
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //1.可以拿到token中的确认结果 结果是true表示发送结束,但不代表发送成功
        boolean ack=token.isComplete();//ack true表示发给服务端了 ack false表示没有发给服务端
        //如果发送是失败 token里会生成一个异常对象
        MqttException exception = token.getException();
        //可以用上述2个数据 才能判断消息是否发送成功
        if (ack&&exception==null){
            log.info("消息发送成功");
        }else{
            log.error("消息发送失败,原因:{}",exception);
        }
    }
}
